use futures::StreamExt;
use log::error;
use paho_mqtt::*;
use serde::{Deserialize, Serialize};
use tokio::sync;

use crate::{Error, NodeId, Result};

/// 管理端通信代理.
///
/// 基于mqtt协议.
/// 监听包含自身节点id的topic,响应命令消息.
pub struct Proxy {
    client: async_client::AsyncClient,
    command_rx: sync::watch::Receiver<NodeCommand>,
    http_client: hyper::Client<hyper::client::connect::HttpConnector>,
}

impl Proxy {
    /// 连接到broker并启动监听.
    pub async fn connect(uri: &str, node_id: &NodeId) -> Result<Self> {
        // mqtt客户端
        let mut client = CreateOptionsBuilder::new()
            .mqtt_version(5)
            .server_uri(uri)
            .client_id(format!("sim_node_{}", node_id))
            .create_client()
            .map_err(Error::from)?;

        // 监听消息流,需要在连接前获取
        let mut msg_rx = client.get_stream(16);

        // 指定版本等连接选项
        let conn_opts = ConnectOptionsBuilder::new()
            .mqtt_version(5)
            .clean_start(true)
            .finalize();
        // 连接到broker
        client.connect(conn_opts).await.map_err(Error::from)?;

        // 启动监听指定topic
        let command_topic = format!("/task/{}", node_id);
        client
            .subscribe(command_topic.as_str(), 2)
            .await
            .map_err(Error::from)?;

        // 用于传递监听消息,只关心最新消息
        let (command_tx, command_rx) = tokio::sync::watch::channel(NodeCommand::default());

        // 启动用于轮询监听消息的协程
        tokio::spawn(async move {
            while let Some(msg) = msg_rx.next().await {
                if let Some(msg) = msg {
                    if msg.topic() == command_topic {
                        // 解析json
                        match serde_json::from_slice::<NetMsgNodeCommand>(msg.payload()) {
                            Ok(cmd) => {
                                if command_tx.send(cmd.into()).is_err() {
                                    break;
                                }
                            }
                            Err(e) => {
                                error!("{}", e);
                            }
                        }
                    }
                }
            }
        });

        let http_client = hyper::Client::new();

        Ok(Self {
            client,
            command_rx,
            http_client,
        })
    }

    /// 发布节点状态.
    pub async fn publish_node_status(&self, node_id: &NodeId, status: &NodeStatus) -> Result<()> {
        let status = NetMsgNodeStatus::new(node_id, status);
        let msg = MessageBuilder::new()
            .topic("/state")
            .payload(serde_json::to_vec(&status).map_err(Error::from)?)
            .qos(2)
            .retained(true)
            .finalize();
        self.client.publish(msg).await.map_err(Error::from)?;

        Ok(())
    }

    /// 生成用于获取命令消息的句柄.
    pub fn get_command_watch(&self) -> NodeCommandWatch {
        NodeCommandWatch {
            command_rx: self.command_rx.clone(),
        }
    }

    /// 从http链接下载.
    ///
    /// 返回字符串形式的内容.
    pub async fn download(&mut self, url: &str) -> Result<Vec<u8>> {
        let req = hyper::Request::get(url)
            .header("User-Agent", "hyper")
            .header("Accept", "*/*")
            .body(hyper::Body::empty())
            .map_err(Error::from)?;
        let res = self.http_client.request(req).await.map_err(Error::from)?;

        if res.status() == hyper::http::StatusCode::OK {
            let content = hyper::body::to_bytes(res.into_body())
                .await
                .map_err(Error::from)?;
            Ok(content.to_vec())
        } else {
            Err(Error::new(format!(
                "Get {} return code: {}",
                url,
                res.status().as_u16()
            )))
        }
    }
}

/// 用于获取命令消息的句柄.
pub struct NodeCommandWatch {
    command_rx: sync::watch::Receiver<NodeCommand>,
}

impl NodeCommandWatch {
    /// 获取命令消息.
    pub async fn get_command(mut self) -> Result<(Self, NodeCommand)> {
        self.command_rx.changed().await.map_err(Error::from)?;
        let cmd = self.command_rx.borrow_and_update().clone();

        Ok((self, cmd))
    }
}

/// 节点状态.
#[derive(Clone, Debug, Serialize)]
pub struct NodeStatus {
    /// 当前已执行的命令id.
    pub command_id: String,
    /// server type.
    pub server_type: String,
    /// 当前加载的地图信息.
    pub map: MapInfo,
    /// 当前任务信息.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub task: Option<TaskInfo>,
    /// 当前场景信息.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scenario: Option<ScenarioInfo>,
    /// 当前车辆列表.
    pub players: Vec<PlayerInfo>,
    /// 客户端连接节点需要的参数.
    pub client_config: ClientConfig,
}

/// 地图信息.
#[derive(Clone, Debug, Serialize)]
pub struct MapInfo {
    /// 名称.
    pub name: String,
    /// 已运行时间.
    pub running_time: u64,
}

/// 任务信息.
#[derive(Clone, Debug, Serialize)]
pub struct TaskInfo {
    /// id.
    pub id: String,
    /// token.
    pub token: String,
}

/// 场景信息.
#[derive(Clone, Debug, Serialize)]
pub struct ScenarioInfo {
    /// id.
    pub id: u64,
    /// 名称.
    pub name: String,
    /// 已运行时间.
    pub running_time: u64,
    /// 是否已停止.
    pub stoped: bool,
    /// 进程是否成功.
    pub proc_success: bool,
    /// 是否通过.
    pub success: bool,
}

/// 车辆信息.
#[derive(Clone, Debug, Serialize)]
pub struct PlayerInfo {
    /// id.
    pub id: String,
    /// 型号.
    pub vehicle_type: String,
    /// 是否为客户端主动创建.
    pub client_created: bool,
    /// 连接到此车辆的客户端id.
    pub client_id: String,
}

/// 客户端连接节点需要的参数.
#[derive(Clone, Debug, Serialize)]
pub struct ClientConfig {
    /// 仿真节点uri,用于zeromq协议.
    pub server_uri: String,
}

/// 节点命令
#[derive(Clone, Debug, Deserialize)]
pub struct NodeCommand {
    /// id.
    pub id: String,
    /// server type(转发到状态).
    pub server_type: String,
    /// 任务设置.
    pub task: Option<TaskSettings>,
    /// 场景设置.
    pub scenario: Option<ScenarioSettings>,
    /// 地图设置.
    pub map: Option<MapSettings>,
}

impl From<NetMsgNodeCommand> for NodeCommand {
    fn from(msg: NetMsgNodeCommand) -> Self {
        let has_task = msg.mode == "start" || msg.mode == "restart";

        Self {
            id: msg.id.clone(),
            server_type: msg.server_type,
            task: if has_task {
                Some(TaskSettings {
                    id: msg.id,
                    token: msg.token,
                })
            } else {
                None
            },
            scenario: if has_task {
                Some(ScenarioSettings {
                    name: msg.scenario.name.unwrap_or_else(|| "".into()),
                    url: msg.scenario.url,
                    is_openscenario: false,
                    is_package: false,
                    force_reload: true,
                })
            } else {
                None
            },
            map: None,
        }
    }
}

/// 任务设置.
#[derive(Clone, Debug, Deserialize)]
pub struct TaskSettings {
    /// id.
    pub id: String,
    /// token(转发到状态).
    pub token: String,
}

/// 场景设置.
#[derive(Clone, Debug, Deserialize)]
pub struct ScenarioSettings {
    /// 名称.
    pub name: String,
    /// 场景描述文件url.
    pub url: String,
    /// 是否为openscenario格式
    pub is_openscenario: bool,
    /// 是否是压缩包
    pub is_package: bool,
    /// 强制重新下载和载入场景文件,否则判断名称是否相同来决定是否重载.
    pub force_reload: bool,
}

/// 地图设置.
#[derive(Clone, Debug, Deserialize)]
pub struct MapSettings {
    /// 名称.
    pub name: String,
    /// 地图描述文件url.
    pub url: String,
    /// 强制重新下载和载入地图文件,否则判断名称是否相同来决定是否重载.
    pub force_reload: bool,
}

impl NodeCommand {
    fn default() -> Self {
        Self {
            id: String::new(),
            server_type: String::new(),
            task: None,
            scenario: None,
            map: None,
        }
    }
}

/// 当前使用的下发命令的json报文.
///
/// 需要转换到内部格式NodeCommand.
#[derive(Clone, Debug, Deserialize)]
struct NetMsgNodeCommand {
    id: String,
    token: String,
    name: Option<String>,
    desc: Option<String>,
    vehicle_type: Option<String>,
    server_type: String,
    mode: String,
    scenario: NetMsgScenario,
}

#[derive(Clone, Debug, Deserialize)]
struct NetMsgScenario {
    id: String,
    name: Option<String>,
    url: String,
}

/// 当前使用的上报状态的json报文.
///
/// 需要从内部格式NodeStatus转换.
#[derive(Clone, Debug, Deserialize, Serialize)]
struct NetMsgNodeStatus {
    id: String,
    server_type: String,
    state: String,
    addr: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    task: Option<NetMsgTaskStatus>,
    message: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
struct NetMsgTaskStatus {
    id: String,
    token: String,
    state: String,
    message: String,
    result: NetMsgScenarioResult,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
struct NetMsgScenarioResult {
    success: bool,
}

impl NetMsgNodeStatus {
    fn new(node_id: &NodeId, status: &NodeStatus) -> Self {
        Self {
            id: format!("{}", node_id),
            // 固定值
            server_type: "carla".into(),
            state: {
                if let Some(scenario) = &status.scenario {
                    if !scenario.stoped {
                        "RUN"
                    } else {
                        "READY"
                    }
                } else {
                    "READY"
                }
            }
            .into(),
            addr: status.client_config.server_uri.clone(),
            task: status.task.as_ref().map(|task| NetMsgTaskStatus {
                id: task.id.clone(),
                token: task.token.clone(),
                state: {
                    if let Some(scenario) = &status.scenario {
                        if !scenario.stoped {
                            "RUN"
                        } else if scenario.proc_success {
                            "DONE"
                        } else {
                            "ERROR"
                        }
                    } else {
                        "ERROR"
                    }
                }
                .into(),
                // TODO:
                message: "".into(),
                result: NetMsgScenarioResult {
                    success: if let Some(scenario) = &status.scenario {
                        scenario.success
                    } else {
                        false
                    },
                },
            }),
            message: "".into(),
        }
    }
}
